package consumer

import (
	"amqp"
	"fmt"
	"iot-cloud-server/command"
	"iot-cloud-server/mongodb"
	"sync"

	"encoding/json"
	"strings"
	"time"

	"gopkg.in/mgo.v2/bson"
)

type Consumer struct {
	DB      *mongodb.MongoServer
	conn    *amqp.Connection
	channel *amqp.Channel
	tag     string
	done    chan error
	wg      *sync.WaitGroup
}

func NewConsumer(amqpURI string, db *mongodb.MongoServer, wg *sync.WaitGroup) (*Consumer, error) {
	// new Consumer
	c := new(Consumer)
	c.conn = nil
	c.channel = nil
	c.tag = "rabbitmq-consumer"
	c.done = make(chan error)
	c.wg = wg
	c.DB = db

	// connect rabbitmq
	var err error
	fmt.Println("dialing %q", amqpURI)
	c.conn, err = amqp.Dial(amqpURI)
	if err != nil {
		return nil, err
	}

	c.wg.Add(1)
	go func() {
		defer c.wg.Done()
		fmt.Println("closing: %s", <-c.conn.NotifyClose(make(chan *amqp.Error)))
	}()

	// get channel
	fmt.Println("getting Channel")
	c.channel, err = c.conn.Channel()
	if err != nil {
		return nil, err
	}

	return c, nil
}

func (c *Consumer) DeclareAndBindExchangeQueue(exchange, exchangeType, queueName, routingKey string) error {
	var err error
	fmt.Println("declaring Exchange %q", exchange)
	if err = c.channel.ExchangeDeclare(
		exchange,     // name of the exchange
		exchangeType, // type
		false,        // durable
		true,         // delete when complete
		false,        // internal
		false,        // noWait
		nil,          // arguments
	); err != nil {
		return err
	}

	fmt.Println("declaring Queue %q", queueName)
	queue, err := c.channel.QueueDeclare(
		queueName, // name of the queue
		true,      // durable
		false,     // delete when usused
		false,     // exclusive
		false,     // noWait
		nil,       // arguments
	)
	if err != nil {
		return err
	}

	fmt.Println("Queue(%q %d messages, %d consumers), binding to Exchange(routingKey %q)",
		queue.Name, queue.Messages, queue.Consumers, routingKey)

	if err = c.channel.QueueBind(
		queue.Name, // name of the queue
		routingKey, // bindingKey
		exchange,   // sourceExchange
		false,      // noWait
		nil,        // arguments
	); err != nil {
		return err
	}

	fmt.Println("Queue bound to Exchange success")
	return nil
}

func (c *Consumer) Consume(queueName, consumerTag string) error {
	var err error
	fmt.Println("Start Consuming")
	deliveries, err := c.channel.Consume(
		queueName,   // name
		consumerTag, // consumerTag,
		false,       // noAck
		false,       // exclusive
		false,       // noLocal
		false,       // noWait
		nil,         // arguments
	)
	if err != nil {
		return err
	}
	c.wg.Add(1)
	go c.Handle(deliveries, c.done)
	return nil
}

func (c *Consumer) Shutdown() error {
	var err error
	// will close() the deliveries channel
	if err = c.channel.Cancel(c.tag, true); err != nil {
		return err
	}

	if err = c.conn.Close(); err != nil {
		return err
	}

	defer fmt.Println("AMQP shutdown OK")

	// wait for handle() to exit
	return <-c.done
}

func (c *Consumer) Handle(deliveries <-chan amqp.Delivery, done chan error) {
	defer func() {
		if r := recover(); r != nil {
			fmt.Println("Handle[E]", r)
		}
	}()
	defer c.wg.Done()
	for d := range deliveries {
		fmt.Println("tag:", d.DeliveryTag, ",body:", string(d.Body))
		d.Ack(false)

		var str []string
		err := json.Unmarshal(d.Body, &str)
		if err != nil {
			fmt.Println("json.Unmarshal err:", err)
		} else if len(str) > 0 {
			recvMsg := strings.Split(str[0], "@@")

			if len(recvMsg) > 1 {
				userData := &command.UserData{Id_: bson.NewObjectId(),
					UserKey:   recvMsg[0],
					TimeStamp: time.Now().Format("2006-01-02 15:04:05"),
					Message:   recvMsg[1],
				}

				go c.DB.Insert(userData)
			}
		}

	}
	fmt.Println("handle: deliveries channel closed")
	done <- nil
}
